package com.atguigu.flink.exec2;

import com.atguigu.flink.pojo.OrderEvent;
import com.atguigu.flink.pojo.UserBehavior;
import com.atguigu.flink.util.MyUtil;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.windowing.ProcessWindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.EventTimeSessionWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

import java.util.List;

/**
 * Created by Smexy on 2022/11/25
 *
 *      规定： 下单后，必须在30分钟内支付
 *
 *              识别超时订单和异常的订单有哪些?
 *
 *
 *      用枚举的方式，列举订单的所有情况，识别出异常。
 *          a)  下单，支付，时间距离不超过30min   正常
 *          b)  下单                          正常
 *          c)  下单，支付， 支付时间超过30min   异常
 *          d)   支付，丢失了订单数据           异常，对不上
 *
 *
 *      思路:
 *              ①按照订单id分组，分组后，多个不同的orderId，分发到同一个Process
 *                      接下来怎么处理?
 *                          可以使用 键控状态处理。
 *
 *               ②使用session窗口。把同一笔订单的数据放入同一个窗口
 *                          每个key有自己的窗口
 *
 *                     窗口被触发时，有以下可能：
 *                          [订单，支付]： 正常，说明下单和支付之间的间隔不超过30min
 *                          [支付]:  第一种:按照一定是先来订单，再支付的流程。
 *                                          从状态读，没有读到，异常，缺少下单
 *                                  第二种： 超时了，订单在之前的窗口已经到了
 *                                              从状态读，读到
 *                          [订单]:
 *                                  正常。只下单，不支付
 *                                          让订单把自己写入状态
 *
 *              基于前提：
 *                      先下单，后支付。
 *                          数据有序，下单一定在支付之前！
 *
 */
public class Demo6_Order
{
    public static void main(String[] args) {


        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        WatermarkStrategy<OrderEvent> watermarkStrategy = WatermarkStrategy.<OrderEvent>forMonotonousTimestamps()
            .withTimestampAssigner((o, t) -> o.getEventTime() * 1000);

        env.readTextFile("data/OrderLog.csv")
           .map(new MapFunction<String, OrderEvent>()
           {
               @Override
               public OrderEvent map(String value) throws Exception {
                   String[] words = value.split(",");
                   return new OrderEvent(
                       Long.valueOf(words[0]),
                       words[1],
                       words[2],
                       Long.valueOf(words[3])
                   );
               }
           }).setParallelism(1)
           .assignTimestampsAndWatermarks(watermarkStrategy)
           .keyBy(OrderEvent::getOrderId)
           .window(EventTimeSessionWindows.withGap(Time.minutes(30)))
           .process(new ProcessWindowFunction<OrderEvent, String, Long, TimeWindow>()
           {

               private ValueState<OrderEvent> order;

               @Override
               public void open(Configuration parameters) throws Exception {
                   order = getRuntimeContext().getState(new ValueStateDescriptor<OrderEvent>("order", OrderEvent.class));
               }

               //当窗口被触发运算时只执行一次
               @Override
               public void process(Long aLong, Context context, Iterable<OrderEvent> elements, Collector<String> out) throws Exception {

                   List<OrderEvent> events = MyUtil.parseList(elements);

                   if (events.size() == 1){

                       //取出，看是下单还是支付
                       OrderEvent orderEvent = events.get(0);

                       if ("create".equals(orderEvent.getEventType())){

                           order.update(orderEvent);

                       }else{
                           //读取状态，读到，就是超时
                           OrderEvent event = order.value();

                           if (event == null){
                               //读不到，就是缺少下单
                               out.collect(orderEvent.getOrderId() + " 只有pay，缺少create");
                           }else{
                               out.collect(orderEvent.getOrderId() + " 支付超时");
                           }

                       }

                   }

               }
           }).print();



        try {
                    env.execute();
                } catch (Exception e) {
                    e.printStackTrace();
                }



    }
}
